/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.wsinfra.wsrm.impl;

import java.math.BigInteger;

import org.apache.xmlbeans.XmlObject;

import cgl.narada.wsinfra.WsMessageFlow;
import cgl.narada.wsinfra.WsProcessor;
import cgl.narada.wsinfra.exception.DeploymentException;
import cgl.narada.wsinfra.exception.IncorrectExchangeException;
import cgl.narada.wsinfra.exception.MessageFlowException;
import cgl.narada.wsinfra.exception.ProcessingException;
import cgl.narada.wsinfra.exception.UnknownExchangeException;
import cgl.narada.wsinfra.exception.WsFaultException;
import cgl.narada.wsinfra.schemas.addressing.ActionDocument;
import cgl.narada.wsinfra.schemas.addressing.EndpointReferenceType;
import cgl.narada.wsinfra.schemas.addressing.FromDocument;
import cgl.narada.wsinfra.schemas.addressing.MessageIDDocument;
import cgl.narada.wsinfra.schemas.addressing.RelatesToDocument;
import cgl.narada.wsinfra.schemas.policy.AppliesToDocument;
import cgl.narada.wsinfra.schemas.policy.PolicyAttachmentDocument;
import cgl.narada.wsinfra.schemas.policy.PolicyDocument;
import cgl.narada.wsinfra.schemas.soap.EnvelopeDocument;
import cgl.narada.wsinfra.schemas.utility.AttributedURI;
import cgl.narada.wsinfra.schemas.wsrm.AckRequestedDocument;
import cgl.narada.wsinfra.schemas.wsrm.AckRequestedType;
import cgl.narada.wsinfra.schemas.wsrm.HandshakeType;
import cgl.narada.wsinfra.schemas.wsrm.SequenceDocument;
import cgl.narada.wsinfra.schemas.wsrm.SequenceFaultDocument;
import cgl.narada.wsinfra.schemas.wsrm.SequenceRefDocument;
import cgl.narada.wsinfra.schemas.wsrm.SequenceRefType;
import cgl.narada.wsinfra.schemas.wsrm.SequenceType;
import cgl.narada.wsinfra.schemas.wsrm.TerminateSequenceDocument;
import cgl.narada.wsinfra.util.SoapMessageAlteration;
import cgl.narada.wsinfra.util.UUIDRetrieval;
import cgl.narada.wsinfra.util.XmlContentTransfer;
import cgl.narada.wsinfra.wsa.AddressingHeaders;
import cgl.narada.wsinfra.wsa.ParseWsaHeaders;
import cgl.narada.wsinfra.wsa.WsaEnvelopeCreator;
import cgl.narada.wsinfra.wsa.WsaEprRules;
import cgl.narada.wsinfra.wsa.WsaFaultRules;
import cgl.narada.wsinfra.wsa.WsaProcessingFactory;
import cgl.narada.wsinfra.wsrm.WsrmAckOperations;
import cgl.narada.wsinfra.wsrm.WsrmActions;
import cgl.narada.wsinfra.wsrm.WsrmElementAddition;
import cgl.narada.wsinfra.wsrm.WsrmElementCreation;
import cgl.narada.wsinfra.wsrm.WsrmExchangeInfo;
import cgl.narada.wsinfra.wsrm.WsrmExchangeInfoCreator;
import cgl.narada.wsinfra.wsrm.WsrmFaults;
import cgl.narada.wsinfra.wsrm.WsrmNodeUtils;
import cgl.narada.wsinfra.wsrm.WsrmProcessingFactory;
import cgl.narada.wsinfra.wsrm.WsrmQNames;
import cgl.narada.wsinfra.wsrm.WsrmRequestCreator;
import cgl.narada.wsinfra.wsrm.WsrmResponseCreator;
import cgl.narada.wsinfra.wsrm.WsrmSequencePolicies;
import cgl.narada.wsinfra.wsrm.WsrmSequencePolicyFactory;
import cgl.narada.wsinfra.wsrm.storage.WsrmAuditStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmPolicyStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmProtocolStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmSequenceInfo;
import cgl.narada.wsinfra.wsrm.storage.WsrmSequenceInfoStorageOperations;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageEventsFactory;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageException;
import cgl.narada.wsinfra.wsrm.storage.WsrmStorageService;

/**
    This interfaces outlines the functionality of a WSRM sink.

    @author Shrideep Pallickara
    $Date: 2005/07/29 22:41:30 $
    $Revision: 1.8 $
*/


public class WsrmSinkNodeHelper extends WsProcessor {
  private static final WsrmSinkNodeHelper instance = new WsrmSinkNodeHelper();
  
  private EndpointReferenceType sinkEpr;
  private XmlContentTransfer xmlContentTransfer;
  private UUIDRetrieval uuidRetrieval;

  private SoapMessageAlteration soapMessageAlteration;
  private WsrmQNames wsrmQNames;
  private WsrmFaults wsrmFaults;
  private WsrmSequencePolicyFactory wsrmSequencePolicyFactory;  
  private WsrmExchangeInfoCreator wsrmExchangeInfoCreator;
  private WsrmAckOperations wsrmAckOperations;
  private WsrmElementAddition wsrmElementAddition;
  private WsrmElementCreation wsrmElementCreation;
  private WsrmRequestCreator wsrmRequestCreator;
  private WsrmNodeUtils wsrmNodeUtils;
  private WsrmActions wsrmActions;
  private WsrmResponseCreator wsrmResponseCreator;
  private WsrmStorageService wsrmStorageService;
  private WsrmProtocolStorageOperations wsrmProtocolOps;
  private WsrmSequenceInfoStorageOperations wsrmSequenceInfoOps;
  private WsrmPolicyStorageOperations wsrmPolicyOps;
  private WsrmAuditStorageOperations wsrmAuditOps;
  private WsrmStorageEventsFactory storageEventsFactory;

  private WsaFaultRules wsaFaultRules;
  private WsaEprRules wsaEprRules;
  private WsaEnvelopeCreator wsaEnvelopeCreator;
  private ParseWsaHeaders parseWsaHeaders; 
  
  private WsMessageFlow wsMessageFlow;
  private String moduleName = "WsrmSinkNodeHelper: ";
  
  private WsrmSinkNodeHelper() {
    xmlContentTransfer = XmlContentTransfer.getInstance();
    soapMessageAlteration = SoapMessageAlteration.getInstance();
    uuidRetrieval = UUIDRetrieval.getInstance();

    wsrmQNames =  WsrmQNames.getInstance();
    wsrmFaults = WsrmProcessingFactory.getWsrmFaults();
    wsrmSequencePolicyFactory = 
      WsrmProcessingFactory.getWsrmSequencePolicyFactory();  
    wsrmExchangeInfoCreator = 
      WsrmProcessingFactory.getWsrmExchangeInfoCreator();
    wsrmAckOperations = WsrmProcessingFactory.getWsrmAckOperations();
    wsrmElementAddition = WsrmProcessingFactory.getWsrmElementAddition();
    wsrmElementCreation = WsrmProcessingFactory.getWsrmElementCreation();
    wsrmRequestCreator = WsrmProcessingFactory.getWsrmRequestCreator();
    wsrmNodeUtils = WsrmProcessingFactory.getWsrmNodeUtils();
    wsrmResponseCreator = WsrmProcessingFactory.getWsrmResponseCreator();
    wsrmActions = WsrmActions.getInstance();
    storageEventsFactory = WsrmStorageEventsFactory.getInstance();
    
    wsaFaultRules =  WsaProcessingFactory.getWsaFaultRules();
    wsaEprRules = WsaProcessingFactory.getWsaEprRules();
    wsaEnvelopeCreator = WsaProcessingFactory.getWsaEnvelopeCreator();
    parseWsaHeaders = WsaProcessingFactory.getParseWsaHeaders();
  }

  
  public static WsrmSinkNodeHelper getInstance() {
    return instance;
  }

  public boolean 
  processExchange(EnvelopeDocument envelopeDocument,  int direction)
    throws UnknownExchangeException, IncorrectExchangeException,
    ProcessingException, MessageFlowException {

    String errorReport = null;

    errorReport = moduleName + "Should NOT have this exchange. This is "
      + "just a helper to the WsrmSinkNode. ";
   
    if (errorReport != null) {
      throw new ProcessingException(errorReport);
    }

    return false;
  }

  /** Sets the message flow which the processor should use */
  public void setMessageFlow(WsMessageFlow wsMessageFlow) 
    throws DeploymentException {
    this.wsMessageFlow = wsMessageFlow;
  }
  
  /** Gets the message flow which the processor should use. */
  public WsMessageFlow getMessageFlow() {
    return wsMessageFlow;
  }

  public void initialize(WsrmStorageService wsrmStorageService) 
    throws DeploymentException {
    this.wsrmStorageService = wsrmStorageService;
    if (wsrmStorageService == null) {
      throw new DeploymentException(moduleName + "StorageService is NULL!");
    }
    
    try {
      wsrmProtocolOps = wsrmStorageService.getWsrmProtocolStorageOperations();
      wsrmSequenceInfoOps = 
	wsrmStorageService.getWsrmSequenceInfoStorageOperations(); 
      wsrmPolicyOps = wsrmStorageService.getWsrmPolicyStorageOperations();
      wsrmAuditOps =  wsrmStorageService.getWsrmAuditStorageOperations();
      
    } catch (WsrmStorageException wsrmStorageException) {
      String reason = moduleName + "Problems initializing storage services\n"
	+ wsrmStorageException.toString();
      throw new DeploymentException(reason);
    }
  }

  
   
  /* Retrieves the endpoint reference associated with this node */
  public EndpointReferenceType getEndpointReference() {
    return sinkEpr;
  }


  /** Retrieves the endpoint reference associated with this node */
  public void setEndpointReference(EndpointReferenceType endpointReference) {
    this.sinkEpr = endpointReference;
  }


  /** Prepares a create sequence response based on the addressing headers
      contained in the original request. This method performs the following
      functions.
      <br>(a) Checks to see if the required elements [wsa:To] & [wsa:MessageID]
      are present. If they are not a ProcessingException or FaultException
      is thrown respectively.
      <br>(b) Prepare the SOAP enevelope with the approporiate destination,
      [wsa:Action] and [wsa:RelatesTo] fields. 
      <br>(c) Create a new sequenceIdentifier, proceed to add the appropriate
      CreateSequenceResponse document based on this.
      <br>(d) Attach policy information to the envelope. 
      <br>(e) Create the wsrmSequenceInfo element and add it to the database.*/
  public EnvelopeDocument 
  prepareCreateSequenceResponseEnvelope(AddressingHeaders addressingHeaders) 
    throws  WsFaultException, ProcessingException, WsrmStorageException {
    String errorReport = null;
    EndpointReferenceType to = wsaEprRules.sendTo(addressingHeaders);
    
    if (to == null) {
      errorReport = moduleName + "Received a CreateSequence request but NO " +
	"wsa:From or wsa:ReplyTo elements are present";
      throw new ProcessingException(errorReport);
    }
    
    FromDocument from = FromDocument.Factory.newInstance();
    from.setFrom(sinkEpr);
    
    ActionDocument action = ActionDocument.Factory.newInstance();
    String actionString = wsrmActions.getCreateSequenceResponse();
    action.addNewAction().setStringValue(actionString);
    
    if (!addressingHeaders.hasMessageID()) {
      /* Report problems here */
      errorReport = " The wsa:MessageID field is missing in the request.";
      throwCreateSequenceRefusedFaultException(addressingHeaders, errorReport);
    }
    
    MessageIDDocument messageID = addressingHeaders.getMessageID();
    String relatesToId = messageID.getMessageID().getStringValue();
    RelatesToDocument relatesTo = RelatesToDocument.Factory.newInstance();
    relatesTo.addNewRelatesTo().setStringValue(relatesToId);
    
    EnvelopeDocument envelopeDocument = 
      wsaEnvelopeCreator.createSoapEnvelope(to, from, action, relatesTo);
    
    /** Create a sequence identifier, and add the create sequence document
	to the newly created envelope */
    String sequenceIdentifier = uuidRetrieval.getRandomBasedUUIDAsString();
    wsrmResponseCreator.addCreateSequenceResponse(envelopeDocument, 
						  sequenceIdentifier);
    WsrmSequencePolicies wsrmSequencePolicies = 
      addPolicyAttachment(envelopeDocument, sequenceIdentifier);
    
    
    /** Create a WsrmSequenceInfo object and store it */
    boolean isSource = false;
    WsrmSequenceInfo wsrmSequenceInfo =
      storageEventsFactory.createWsrmSequenceInfo(sequenceIdentifier,
						  relatesToId,
						  to, sinkEpr, isSource);
    storageEventsFactory.attachPolicyToSequence(wsrmSequenceInfo, 
						wsrmSequencePolicies);
    wsrmSequenceInfoOps.storeCreatedSequence(wsrmSequenceInfo);
    
    return envelopeDocument;
  }
  
  
  /** Adds a policyAttachment to the header of the provide envelope. Make sure
      that a header is present within the supplied envelope. This attachment
      is typically added to the header of the CreateSequenceResponse envelope
      document.*/ 
  public  WsrmSequencePolicies
  addPolicyAttachment(EnvelopeDocument envelopeDocument,
		      String sequenceIdentifier) {
    PolicyAttachmentDocument policyAttachmentDocument =
      PolicyAttachmentDocument.Factory.newInstance();
    PolicyAttachmentDocument.PolicyAttachment policyAttachment = 
      policyAttachmentDocument.addNewPolicyAttachment();

    AppliesToDocument.AppliesTo appliesTo = policyAttachment.addNewAppliesTo();

    SequenceRefDocument sequenceRefDocument = 
      SequenceRefDocument.Factory.newInstance();
    SequenceRefType sequenceRef = sequenceRefDocument.addNewSequenceRef();
    sequenceRef.setIdentifier(sequenceIdentifier);
    sequenceRef.setMatch(WsrmQNames.getInstance().getExact());

    xmlContentTransfer.copyFromSourceToDestination(sequenceRefDocument, 
						   appliesTo);
    
    WsrmSequencePolicies wsrmSequencePolicies = 
      wsrmSequencePolicyFactory.getDefaultWsrmSequencePolicies();

    PolicyDocument.Policy policy = policyAttachment.addNewPolicy();
    addWsrmPolicy(policy, wsrmSequencePolicies.getAcknowledgementInterval());
    addWsrmPolicy(policy, wsrmSequencePolicies.getRetransmissionInterval() );
    addWsrmPolicy(policy, wsrmSequencePolicies.getExponentialBackoff() );
    addWsrmPolicy(policy, wsrmSequencePolicies.getInactivityTimeout() );
    
    soapMessageAlteration.addToSoapHeader(envelopeDocument, 
					  policyAttachmentDocument);

    return wsrmSequencePolicies;
  }
  
  
  /** Adds the wsrm policy object to the specified policy */
  private void 
  addWsrmPolicy(PolicyDocument.Policy policy, XmlObject wsrmPolicyObject) {
    if (policy == null || wsrmPolicyObject== null) {
      System.out.println(moduleName + "Trying to add a NULL WSRM policy");
      return;
    }
    xmlContentTransfer.copyFromSourceToDestination(wsrmPolicyObject, policy);
  }




  /** Upon receipt of a sequence this method checks to see if any problems
      exist within the element. Specifically we check for the following
      problems
      <br>(a)Make sure that there is a wsu:Identifier element in Sequence, and
      that it is not of a NULL value.
      <br>(b)Make sure that we know about the sequence.
      <br>(c)Make sure that this sequence is not terminated.
      <br>(d)Check to see that the message number is NOT ZERO.
      <br>(e)Check to see that the LastMessageNumber constraint is not
      exceeded.
      <br>(f)Check to see that the source is not trying to OVERRIDE a 
      PREVIOUSLY established last message number. This method returns the
      WsrmSequenceInfo corresponding to the element; this is done to 
      reduce the number of database accesses. */
  public WsrmSequenceInfo
  checkSequenceElementForProblems(AddressingHeaders addressingHeaders,
				  SequenceDocument sequenceDocument) 
    throws WsrmStorageException, WsFaultException {
    String errorReport = "";
    String enclosingElement = "[wsrm:Sequence]";
    SequenceType sequenceType = sequenceDocument.getSequence();
    AttributedURI attributedUri = sequenceType.getIdentifier();

    checkAttributedURIElementForProblems(addressingHeaders, attributedUri, 
					 enclosingElement);

    
    String sequenceIdentifier = attributedUri.getStringValue();
    WsrmSequenceInfo wsrmSequenceInfo =
      checkSequenceIdentifierForProblems(addressingHeaders, sequenceIdentifier,
					 enclosingElement);

    boolean isLastMessage = sequenceType.isSetLastMessage();
    BigInteger bigMessageNumber = sequenceType.getMessageNumber();
    if (bigMessageNumber == null) {
      errorReport = "There is no message number that has been specified in " +
	"the [wsrm:Sequence] element. Check the message.";
      throwInvalidMessageFaultException(errorReport, addressingHeaders);
    }
    
    long messageNumber = bigMessageNumber.longValue();
    
    /** Check to see if the message number is 0 */
    if (messageNumber == 0) {
      errorReport = "The message number specified in the " + enclosingElement +
	"element is 0. This is not a valid MessageNumber. Check the message.";
      throwInvalidMessageFaultException(errorReport, addressingHeaders);
    }

    /** Check to make sure that the last message info is not exceeded.
	Also make sure that the lastMessageNumber is constant, that is once
	set no one is allowed to change it to something else. */
    long lastMessageNumber = wsrmSequenceInfo.getLastMessageNumber();
    if (wsrmSequenceInfo.hasLastMessageInfo() ) {
      if (messageNumber > lastMessageNumber) {
	errorReport = "The previously established LastMessageNumber=" +
	  lastMessageNumber + " & the exchange has MessageNumber" 
	  + messageNumber;
	throwLastMessageNumberExceededFaultException(addressingHeaders,
						     sequenceIdentifier,
						     errorReport);
      }
      
      if (isLastMessage) {
	if (messageNumber != lastMessageNumber) {
	  errorReport = "The previously established wsrm:LastMessageNumber=["+
	    lastMessageNumber + "], now trying to reset it to [" + 
	    messageNumber + "]. Set the LastMessage ONLY on the last message";
	  throwInvalidMessageFaultException(errorReport, addressingHeaders);
	}
      }
    } /** wsrmSequenceInfo.hasLastMessageInfo() */
    
    /** Proceed to update the sequence information if this is indeed the
	last message of the sequence. */
    if (isLastMessage) {
      storageEventsFactory.setLastMessageNumberOnSequence(wsrmSequenceInfo,
							  messageNumber); 
      wsrmSequenceInfoOps.storeCreatedSequence(wsrmSequenceInfo);
    }
    
    return wsrmSequenceInfo;
  }





  /** Checks to see if there is already a widget that has been stored
      corresponding to messageNumber and sequenceIdentifier encapsulated within
      the [wsrm:SequenceDocument]. */
  public boolean
  checkIfDuplicate(SequenceDocument sequenceDocument) 
    throws WsrmStorageException {
    boolean duplicate = false;

    SequenceType sequenceType = sequenceDocument.getSequence();
    String sequenceIdentifier = sequenceType.getIdentifier().getStringValue();
    long messageNumber = sequenceType.getMessageNumber().longValue();
    
    duplicate = wsrmProtocolOps.hasStoredElement(sequenceIdentifier, 
						 messageNumber);
    return duplicate;
  }


  
  /** Upon receipt of a sequence this method checks to see if any problems
      exist within the element. Specifically we check for the following
      problems
      <br>(a)Make sure that there is a wsu:Identifier element in Sequence, and
      that it is not of a NULL value.
      <br>(b)Make sure that we know about the sequence.
      <br>(c)Make sure that this sequence is not terminated.
      <br>(d)Check to see that the message number is NOT ZERO.
      <br>(e)Check to see that the LastMessageNumber constraint is not
      exceeded. */
  public WsrmSequenceInfo
  checkAckRequestedElementForProblems(AddressingHeaders addressingHeaders,
				      AckRequestedDocument ackRequestedDoc) 
  throws WsrmStorageException, WsFaultException {
    AckRequestedType ackRequested = ackRequestedDoc.getAckRequested();
    String enclosingElement = "[wsrm:AckRequested]";
    String errorReport = null;
    AttributedURI attributedUri = ackRequested.getIdentifier();

    checkAttributedURIElementForProblems(addressingHeaders, attributedUri, 
					 enclosingElement);

    
    String sequenceIdentifier = attributedUri.getStringValue();
    WsrmSequenceInfo wsrmSequenceInfo =
      checkSequenceIdentifierForProblems(addressingHeaders, sequenceIdentifier,
					 enclosingElement);
    
    if (ackRequested.isSetMaxMessageNumberUsed() ) {
      long maxMessageNumberUsed = 
	ackRequested.getMaxMessageNumberUsed().longValue();

      if (maxMessageNumberUsed == 0) {
	errorReport = "The message number specified in the " + 
	  enclosingElement + "element is 0. This is not a valid MessageNumber."
	  + " Check the message.";
	throwInvalidMessageFaultException(errorReport, addressingHeaders);
      }
      
      long lastMessageNumber = wsrmSequenceInfo.getLastMessageNumber();
      if (wsrmSequenceInfo.hasLastMessageInfo() ) {
	if (maxMessageNumberUsed > lastMessageNumber) {
	  errorReport = "The previously established LastMessageNumber=" +
	    lastMessageNumber + " & the " + enclosingElement + 
	    " exchange has MaxMessageNumber="    + maxMessageNumberUsed;
	  throwLastMessageNumberExceededFaultException(addressingHeaders,
						       sequenceIdentifier,
						       errorReport);
	}
      }
    }
    
    return wsrmSequenceInfo;
  }




  /** Prepares a SOAP envelope for the acknowledgement based on the
      specified addressingHeaders (contained within an AckRequested element)
      and the SequenceIdentifier contained within the WsrmSequenceInfo. */
  public EnvelopeDocument 
  prepareEnvelopeForAcknowledgement(AddressingHeaders addressingHeaders,
				    WsrmSequenceInfo wsrmSequenceInfo) 
    throws  WsFaultException, ProcessingException {
    String errorReport = null;
    EndpointReferenceType to = wsaEprRules.sendTo(addressingHeaders);
    
    if (to == null) {
      errorReport = moduleName + "Received a CreateSequence request but NO " +
	"wsa:From or wsa:ReplyTo elements are present";
      throw new ProcessingException(errorReport);
    }
    
    FromDocument from = FromDocument.Factory.newInstance();
    from.setFrom(sinkEpr);
    
    ActionDocument action = null;
    
    if (!addressingHeaders.hasMessageID()) {
      /* Report problems here */
      errorReport = " The wsa:MessageID field is missing in the request.";
      throwCreateSequenceRefusedFaultException(addressingHeaders, errorReport);
    }
    
    
    String relatesToId = wsrmSequenceInfo.getAddressingIdentifier();
    RelatesToDocument relatesTo = RelatesToDocument.Factory.newInstance();
    relatesTo.addNewRelatesTo().setStringValue(relatesToId);
    
    EnvelopeDocument envelopeDocument = 
      wsaEnvelopeCreator.createSoapEnvelope(to, from, action, relatesTo);
    
    return envelopeDocument;
  }



  
  
  /** Upon receipt of a sequence this method checks to see if any problems
      exist within the element. Specifically we check for the following
      problems
      <br>(a)Make sure that there is a wsu:Identifier element in Sequence, and
      that it is not of a NULL value.
      <br>(b)Make sure that we know about the sequence.
      <br>(c)Make sure that this sequence is not terminated. 
      <br>(d)Make sure that the sequence being terminated DOES NOT have any
      pending acknowledgements to issue.*/
  public WsrmSequenceInfo
  checkTerminateSequenceElementForProblems(AddressingHeaders addressingHeaders,
					   TerminateSequenceDocument 
					   terminateSequenceDocument) 
    throws WsrmStorageException, WsFaultException {
    HandshakeType terminateSequence = 
      terminateSequenceDocument.getTerminateSequence();
    String enclosingElement = "[wsrm:TerminateSequence]";
    String errorReport = null;
    AttributedURI attributedUri = terminateSequence.getIdentifier();
    
    checkAttributedURIElementForProblems(addressingHeaders, attributedUri, 
					 enclosingElement);
    
    
    String sequenceIdentifier = attributedUri.getStringValue();
    WsrmSequenceInfo wsrmSequenceInfo =
      checkSequenceIdentifierForProblems(addressingHeaders, sequenceIdentifier,
					 enclosingElement);
    
    /** Check to see if there are pending acknowledgements on the sequence */
    boolean ackRequested = true; //over-ride the ackInterval constraint.
    long[] pendingAcks = 
      wsrmProtocolOps.getUnacknowledgedMessageNumbers(sequenceIdentifier, 
						      ackRequested);
    
    if (pendingAcks != null) {
      errorReport = "There are (" + pendingAcks.length + 
	") pending acknowledgements for the Sequence [" + sequenceIdentifier +
	"]. One cannot terminate Sequences with pending acknowledgements.";
      throwInvalidMessageFaultException(errorReport, addressingHeaders);
    }

    return wsrmSequenceInfo;
  }




  /** <br>(b)Make sure that we know about the sequence.
      <br>(c)Make sure that this sequence is not terminated. */
  private WsrmSequenceInfo 
  checkSequenceIdentifierForProblems(AddressingHeaders addressingHeaders,
				     String sequenceIdentifier,
				     String enclosingElement) 
    throws WsrmStorageException, WsFaultException {
    String errorReport = null;
    if (!wsrmSequenceInfoOps.isKnownSequence(sequenceIdentifier) ) {
      throwUnknownSequenceFaultException(addressingHeaders, 
					 sequenceIdentifier);
    }
    
    WsrmSequenceInfo wsrmSequenceInfo =
      wsrmSequenceInfoOps.getSequenceInfoUsingSequenceIdentifier(sequenceIdentifier);
    
    if (wsrmSequenceInfo.isTerminated()) {
      errorReport = "The specified sequence (" + sequenceIdentifier + 
	") in " + enclosingElement + "] has been [Terminated] because " + 
	wsrmSequenceInfo.getTerminationReason(); 
      throwSequenceTerminatedFaultException(addressingHeaders, 
					    sequenceIdentifier, errorReport);
    }
    
    return wsrmSequenceInfo;
  }

  
  /** Make sure that there is a wsu:Identifier element in Sequence, and
      that it is not of a NULL value.*/
  private void 
  checkAttributedURIElementForProblems(AddressingHeaders addressingHeaders, 
				       AttributedURI attributedUri, 
				       String enclosingElement) 
    throws WsFaultException {
    String errorReport = null;
    if (attributedUri == null) {
      errorReport = moduleName + "There is NO wsu:Identifier element in the " +
	enclosingElement + " element. Check the created message.";
      throwInvalidMessageFaultException(errorReport, addressingHeaders);
    }
  }



  /** Throws a WsFaultException corresponding to the LastMessageNumberExceeded
      fault. */
  public void 
  throwLastMessageNumberExceededFaultException(AddressingHeaders 
					       addressingHeaders,
					       String sequenceIdentifier,
					       String additionalReason) 
    throws WsFaultException {

    EndpointReferenceType faultTo = 
      wsaFaultRules.sendFaultTo(addressingHeaders);
    
    String reason = wsrmFaults.getLastMessageNumberExceededReason();
    if (additionalReason != null) {
      reason += additionalReason;
    }

    WsFaultException wsFaultException = 
      new WsFaultException(wsrmFaults.getSoapSenderQName(), reason );
    
    SequenceFaultDocument sequenceFaultDocument = 
      wsrmFaults.getLastMessageNumberExceeded(sequenceIdentifier);
    
    wsFaultException.setAddSoapHeaderElement( sequenceFaultDocument );
    wsFaultException.setFaultTo(faultTo);
    
    throw wsFaultException;
  }


  


  /** Throws a WsFaultException corresponding to the sequence terminated
      fault. */
  public void
  throwSequenceTerminatedFaultException(AddressingHeaders addressingHeaders,
					String sequenceIdentifier,
					String additionalReason) 
    throws WsFaultException {
    EndpointReferenceType faultTo = 
      wsaFaultRules.sendFaultTo(addressingHeaders);
    
    String reason = wsrmFaults.getSequenceTerminatedReason();
    if (additionalReason != null) {
      reason += additionalReason;
    }

    WsFaultException wsFaultException = 
      new WsFaultException(wsrmFaults.getSoapSenderQName(), reason );
    
    SequenceFaultDocument sequenceFaultDocument = 
      wsrmFaults.getSequenceTerminated(sequenceIdentifier);
    
    wsFaultException.setAddSoapHeaderElement( sequenceFaultDocument );
    wsFaultException.setFaultTo(faultTo);
    
    throw wsFaultException;
  }






  /** Throws a CreateSequenceRefused Fault exception, if additional reason
      is not null, it will be added to the default reason element. */
  public void 
  throwCreateSequenceRefusedFaultException(AddressingHeaders addressingHeaders,
					   String additionalReason) 
  throws WsFaultException {
    EndpointReferenceType faultTo = 
      wsaFaultRules.sendFaultTo(addressingHeaders);
    
    String reason = wsrmFaults.getCreateSequenceRefusedReason();
    if (additionalReason != null) {
      reason += additionalReason;
    }

    WsFaultException wsFaultException = 
      new WsFaultException(wsrmFaults.getSoapSenderQName(), reason );
    
    SequenceFaultDocument sequenceFaultDocument = 
      wsrmFaults.getCreateSequenceRefused();
    
    wsFaultException.setAddSoapHeaderElement( sequenceFaultDocument );
    wsFaultException.setFaultTo(faultTo);
    
    throw wsFaultException;
  }


  /** Throws an Invalid MessageFault exception based on the specified
      parameters */
  public void 
  throwInvalidMessageFaultException(String reason, 
				    AddressingHeaders addressingHeaders) 
    throws WsFaultException {
    EndpointReferenceType faultTo = 
      wsaFaultRules.sendFaultTo(addressingHeaders);
    
    WsFaultException wsFaultException = 
      new WsFaultException(wsrmFaults.getSoapSenderQName(),
			   reason);
    SequenceFaultDocument sequenceFaultDocument = 
      wsrmFaults.getInvalidMessage();

    wsFaultException.setAddSoapHeaderElement( sequenceFaultDocument );
    wsFaultException.setFaultTo(faultTo);
    
    throw wsFaultException;
  }
  

  /** Throws an UnknownSequence Fault exception based on the specified
      parameters */
  public void 
  throwUnknownSequenceFaultException(AddressingHeaders addressingHeaders,
				     String identifier) 
    throws WsFaultException {
    EndpointReferenceType faultTo = 
      wsaFaultRules.sendFaultTo(addressingHeaders);
    
    WsFaultException wsFaultException = 
      new WsFaultException(wsrmFaults.getSoapSenderQName(),
			   wsrmFaults.getUnknownSequenceReason() );
    SequenceFaultDocument sequenceFaultDocument = 
      wsrmFaults.getUnknownSequence(identifier);

    wsFaultException.setAddSoapHeaderElement( sequenceFaultDocument );
    wsFaultException.setFaultTo(faultTo);
    
    throw wsFaultException;
  }

  /** Checks an exchange type for validity. If there are problems exceptions
      are thrown. */
  public void checkExchangeType(WsrmExchangeInfo wsrmExchangeInfo, 
				int direction)
    throws UnknownExchangeException, IncorrectExchangeException {
    
    if (direction == WsMessageFlow.FROM_NETWORK) {
      checkExchangeFromNetwork(wsrmExchangeInfo);
    }
    
    if (direction == WsMessageFlow.FROM_APPLICATION) {
      checkExchangeFromApplication(wsrmExchangeInfo);
    }
    
  }

  /** ONLY CreateSequence, TerminateSequence, Sequence & AckRequested 
      valid exchanges that should be routed here. */
  private void 
  checkExchangeFromNetwork(WsrmExchangeInfo wsrmExchangeInfo)
    throws UnknownExchangeException, IncorrectExchangeException {
    
    String from = "Network. ";
    String problemElement = "";
    
    if (wsrmExchangeInfo.isCreateSequenceResponse() ) {
      problemElement += " CreateSequenceResponse";
    }

    if (wsrmExchangeInfo.hasSequenceAcknowledgement() ) {
      problemElement += " SequenceAcknowledgement";
    }

    if (wsrmExchangeInfo.hasSequence() || wsrmExchangeInfo.hasAckRequested() ||
	wsrmExchangeInfo.isCreateSequence() || 
	wsrmExchangeInfo.isTerminateSequence() ) {
      /** If this is true, the exchange is a valid one. Proceed to return */
      return;
    }
    
    if (!problemElement.equals("")) {
      throwIncorrectExchangeException(problemElement, from);
    }
    
    
    
    String shouldHaveReceived = 
      "CreateSequence or TerminateSequence or Sequence or AckRequested";
    throwUnknownExchangeException(shouldHaveReceived, from);
  }
  
  
  /** Only CreateSequence and WsrmMessages (messages without action elements
      or WSRM element) are valid. */
  private void 
  checkExchangeFromApplication(WsrmExchangeInfo wsrmExchangeInfo)
    throws UnknownExchangeException, IncorrectExchangeException {
    
    String from = "Application. ";
    String problemElement = "";
    if (wsrmExchangeInfo.hasAckRequested() ) {
      problemElement += " AckRequested ";
    }
    
    if (wsrmExchangeInfo.hasSequence()) {
      problemElement += " Sequence ";
    }
    
    if (wsrmExchangeInfo.hasSequenceAcknowledgement() ) {
      problemElement += " SequenceAcknowledgement ";
    }
    
    if (wsrmExchangeInfo.isCreateSequence() ) {
      problemElement += " CreateSequence ";
    }
    
    if (wsrmExchangeInfo.isTerminateSequence() ) {
      problemElement += " TerminateSequence ";
    }
    
    if (!problemElement.equals("")) {
      throwIncorrectExchangeException(problemElement, from);
    }

  }


  private void throwIncorrectExchangeException(String element, String from)
    throws IncorrectExchangeException {
    String reason =  moduleName +
      "Should NOT have received exchange with element [" +
      element + "] from the " + from;
    throw new IncorrectExchangeException(reason);
  }


  private void throwUnknownExchangeException(String element, String from)
    throws UnknownExchangeException {
    String reason =  moduleName +
      "No idea about processing exchange without elements [" + element + 
      "] from the " + from + ".";
    throw new UnknownExchangeException(reason);
  }

}
